package com.ayz.gateway.client;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.log.Log;

import com.ayz.gateway.api.request.CheckServer;
import com.ayz.gateway.api.response.GatewayCommonResult;
import com.ayz.gateway.api.server.CheckServerServiceGrpc;
import com.ayz.gateway.client.connection.Connection;
import com.ayz.gateway.client.connection.GrpcConnection;
import com.ayz.gateway.common.result.ResponseCode;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

import java.util.List;
import java.util.concurrent.*;

/**
 * @ClassName GrpcClient
 * @Description TODO: 基于GRPC实现的 远程通信的服务客户端
 * @Author 快乐的星球
 * @Date 2023/5/21 16:23
 * @Version 1.0
 **/
public class GrpcClient extends RpcClient{
    private static final Log LOG=Log.get(GrpcClient.class);
    /**
     *  GRPC客户端的Netty使用的线程池
     */
    private Executor grpcExecutor;

    private GrpcClientConfig grpcClientConfig;

    public GrpcClient(List<String> serverListAddress, GrpcClientConfig rpcClientConfig) {
        super(serverListAddress, rpcClientConfig);
        this.grpcClientConfig=rpcClientConfig;
    }
    public GrpcClient(GrpcClientConfig rpcClientConfig) {
        super(rpcClientConfig);
        this.grpcClientConfig=rpcClientConfig;
    }

    @Override
    public Connection connectToServer(String serverAddress) {
        //1: 构建ServerInfo对象
        ServerInfo serverInfo = ServerInfo.getInstance(serverAddress);
        //2:创建GRPC客户端连接的MangedChannel
        ManagedChannel newMangedChannel = createNewMangedChannel(serverInfo.getIp(), serverInfo.getPort());
        if(newMangedChannel==null){
            throw new RuntimeException("创建MangedChannel失败!");
        }
        //3:测试当前客户端连接是否能够与服务器双向通信
        boolean checkSuccess=true;
        try {
            GatewayCommonResult.GatewayResult gatewayResult = checkServer(newMangedChannel);
            int resultCode = gatewayResult.getCode();
            if(resultCode!= ResponseCode.COMMON_SUCCESS){
                checkSuccess=false;
                String msg = gatewayResult.getMsg();
                LOG.error("向指定服务器:{"+serverAddress+"} 发起CheckServerRequest请求失败!错误信息为:{"+msg+"}");
            }
        } catch (ExecutionException e) {
            LOG.warn("向服务器:{"+serverAddress+"} 发CheckServerRequest请求失败!");
            checkSuccess=false;
        } catch (InterruptedException e) {
            checkSuccess=false;
        } catch (TimeoutException e) {
            LOG.warn("向服务器:{"+serverAddress+"} 发CheckServerRequest请求超时!");
            checkSuccess=false;
        }
        //检查成功则创建Connection 否则返回 空值
        if(checkSuccess){
            return new GrpcConnection(newMangedChannel,serverInfo,getRequestExecutorRegistry());
        }else {
            //检查失败
            shutdownChannel(newMangedChannel);
            return null;
        }
    }

    @Override
    public void registryRequestExecutors(RequestExecutorRegistryCallback registryCallback) {
        /**
         * TODO: 完成请求执行器的注册
         */
        registryCallback.callback(getRequestExecutorRegistry());
    }

    /**
     * 与指定的服务端建立起GRPC长连接
     * @param serverIp
     * @param port
     * @return
     */
    private ManagedChannel createNewMangedChannel(String serverIp,int port){
        this.grpcExecutor=createNewExecutor(serverIp);
        //构建MangedChannel
        ManagedChannelBuilder<?> managedChannelBuilder=ManagedChannelBuilder.forAddress(serverIp,port)
                .executor(grpcExecutor).compressorRegistry(CompressorRegistry.getDefaultInstance())
                .decompressorRegistry(DecompressorRegistry.getDefaultInstance()).maxInboundMessageSize(getGrpcClientConfig().getMaxInboundMessageSize())
                .keepAliveTime(getGrpcClientConfig().getChannelKeepAlive(),TimeUnit.MILLISECONDS)
                .keepAliveTimeout(getGrpcClientConfig().getChannelKeepAliveTimeout(), TimeUnit.MILLISECONDS)
                .usePlaintext();

        return managedChannelBuilder.build();
    }

    /**
     * 关闭一个MangedChannel
     * @param managedChannel
     */
    private void shutdownChannel(ManagedChannel managedChannel){
        if(managedChannel!=null && !managedChannel.isShutdown()){
            managedChannel.shutdown();
        }
    }
    /**
     * GRPC客户端依赖于Netty，当前线程池是为Netty客户端所使用
     * @param serverIp
     * @return
     */
    private ThreadPoolExecutor createNewExecutor(String serverIp){
        GrpcClientConfig grpcClientConfig = getGrpcClientConfig();
        return new ThreadPoolExecutor(grpcClientConfig.getThreadPoolCoreSize(),grpcClientConfig.getThreadPoolMaxSize(),grpcClientConfig.getThreadPoolKeepAlive(),
                TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(grpcClientConfig.getThreadPoolQueueSize()),
                new ThreadFactoryBuilder().setDaemon(true).setNamePrefix("gatewayClient:"+serverIp+"_").build());
    }

    /**
     * 当客户端与服务器采用GRPC通信时
     * 建立GRPC客户端ManagedChannel后需要发送一次 CheckServerRequest请求来检测当前连接是否是双向导通的
     * @param managedChannel
     * @return
     */
    public GatewayCommonResult.GatewayResult checkServer(ManagedChannel managedChannel) throws ExecutionException, InterruptedException, TimeoutException {
        //1:准备CheckServerRequest请求
        CheckServer.CheckServerRequest checkServerRequest = CheckServer.CheckServerRequest.newBuilder().build();
        //2:获取存根
        CheckServerServiceGrpc.CheckServerServiceFutureStub checkServerService = CheckServerServiceGrpc.newFutureStub(managedChannel);
        ListenableFuture<GatewayCommonResult.GatewayResult> gatewayResultFuture = checkServerService.checkServer(checkServerRequest);
        return gatewayResultFuture.get(grpcClientConfig.getCheckServerRequestTimeout(),TimeUnit.MILLISECONDS);
    }
    public Executor getGrpcExecutor() {
        return grpcExecutor;
    }

    public void setGrpcExecutor(Executor grpcExecutor) {
        this.grpcExecutor = grpcExecutor;
    }

    public GrpcClientConfig getGrpcClientConfig() {
        return grpcClientConfig;
    }

    public void setGrpcClientConfig(GrpcClientConfig grpcClientConfig) {
        this.grpcClientConfig = grpcClientConfig;
    }
}
